package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{CollectionAccumulator, DoubleAccumulator, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

object Demo23Acc {
  def main(args: Array[String]): Unit = {

    /**
     * accumulate：累加器
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8), 4)

    // 算子外部的代码是在Driver端执行的
    var cnt: Int = 0
    println(s"job运行前cnt的值为$cnt") // 0

    // 在Driver端初始化累加器，默认值：值类型为0，引用类型为null
    val longAcc: LongAccumulator = sc.longAccumulator
    val doubleAcc: DoubleAccumulator = sc.doubleAccumulator
    val collectAcc: CollectionAccumulator[Int] = sc.collectionAccumulator[Int]

    rdd.map(i => {
      // 算子内部的代码会被封装到Task中由Driver发送到Executor中执行
      // 算子内部如果使用了外部定义的变量，实际上用的是外部变量的副本
      // 在算子内部直接修改变量，实际上修改的是变量副本的值，并不会对外部的变量造成影响
      cnt += 1

      // 在算子内部进行累加
      longAcc.add(1)
      doubleAcc.add(1.0)
      collectAcc.add(i)
      println(s"map任务中的cnt为:$cnt")
      println(s"map任务中的累加器longAcc的值为:${longAcc.value}")
      i
    }).foreach(println)

    println(s"job运行完成后cnt的值为$cnt") // 0
    // 在Driver端进行了汇总，会将每个Executor中的累加器进行合并
    println(s"job运行完成后累加器longAcc的值为:${longAcc.value}") // 8
    println(s"job运行完成后累加器doubleAcc的值为:${doubleAcc.value}") // 8
    println(s"job运行完成后累加器collectAcc的值为:${collectAcc.value}") // [1, 2, 3, 4, 5, 6, 7, 8]


  }

}
